package cn.itcast.flink.process.timer;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * 将数据流保存至MySQL数据库表中
 *
 * @author lilulu
 * @date 2023-04-07 22:51
 */
public class OrderSink extends RichSinkFunction<OrderData> {

    private Connection connection = null;
    private PreparedStatement preparedStatement = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection("jdbc:mysql://node1:3306/?useSSL=false", "root", "123456");
        preparedStatement = connection.prepareStatement("INSERT INTO db_flink.tbl_orders (order_id, user_id, order_time, order_status, order_amount) VALUES (?,?,?,?,?)");
    }


    @Override
    public void invoke(OrderData orderData, Context context) throws Exception {
        preparedStatement.setString(1, orderData.getOrderId());
        preparedStatement.setString(2, orderData.getUserId());
        preparedStatement.setString(3, orderData.getOrderTime());
        preparedStatement.setString(4, orderData.getOrderStatus());
        preparedStatement.setDouble(5, orderData.getOrderAmount());
        preparedStatement.execute();
    }

    @Override
    public void close() throws Exception {
        if (preparedStatement != null && !preparedStatement.isClosed()) {
            preparedStatement.close();
        }
        if (connection != null && !connection.isClosed()) {
            connection.close();
        }
    }

}
